package com.swak.algs.paxos;

import java.util.Queue;

import com.swak.algs.sample.Samples;
import com.swak.utils.Queues;

/**
 * 模拟的网络
 * 
 * @author 618lf
 */
public class NetWork extends Thread {

	private double SendPrepareMessageProbability = 50;
	private double SendPrepareResponseMessageProbability = 50;
	private double SendAcceptMessageProbability = 50;
	private double SendAcceptResponseMessageMessageProbability = 50;

	/**
	 * 任务队列
	 */
	private Queue<Task> tasks = Queues.newArrayBlockingQueue(2000);

	/**
	 * 发送准备消息
	 */
	void sendPrepareMessage(Proposal proposal) {
		Task task = new Task();
		task.v = proposal;
		this.tasks.add(task);
	}

	private void doSendPrepareMessage(Task task) {
		for (int i = 0; i < Paxos.me().getAcceptors().size(); i++) {
			boolean success = Samples.probability(SendPrepareMessageProbability);
			if (success) {
				Acceptor acceptor = Paxos.me().getAcceptors().get(i);
				acceptor.addTask(task);
			}
		}
	}

	/**
	 * 准备消息的响应
	 */
	void sendPrepareResponseMessage(PrepareResult proposalResult) {
		Task task = new Task();
		task.v = proposalResult;
		this.tasks.add(task);
	}

	private void doSendPrepareResponseMessage(PrepareResult proposalResult, Task task) {
		boolean success = Samples.probability(SendPrepareResponseMessageProbability);
		if (success) {
			int proposerId = proposalResult.getProposerId();
			Proposer proposer = Paxos.me().getProposers().get(proposerId);
			proposer.addTask(task);
		}
	}

	/**
	 * 发送第二阶段消息
	 */
	void sendAcceptMessage(Proposal proposal) {
		Task task = new Task();
		task.v = proposal;
		this.tasks.add(task);
	}

	private void doSendAcceptMessage(Task task) {
		for (int i = 0; i < Paxos.me().getAcceptors().size(); i++) {
			boolean success = Samples.probability(SendAcceptMessageProbability);
			if (success) {
				Acceptor acceptor = Paxos.me().getAcceptors().get(i);
				acceptor.addTask(task);
			}

		}
	}

	/**
	 * 发送第二阶段消息 -- 响应
	 */
	void sendAcceptResponseMessage(AcceptResult proposalResult) {
		Task task = new Task();
		task.v = proposalResult;
		this.tasks.add(task);
	}

	private void doSendAcceptResponseMessage(AcceptResult proposalResult, Task task) {
		boolean success = Samples.probability(SendAcceptResponseMessageMessageProbability);
		if (success) {
			int proposerId = proposalResult.getProposerId();
			Proposer proposer = Paxos.me().getProposers().get(proposerId);
			proposer.addTask(task);
		}
	}

	/**
	 * 模拟网络运行
	 */
	@Override
	public void run() {
		while (true) {

			/*
			 * 获取任务
			 */
			Task task = tasks.poll();
			if (task == null) {
				continue;
			}

			Object v = task.v;

			// 准备阶段
			if (v instanceof Proposal && ((Proposal) v).getStep() == 1) {
				this.doSendPrepareMessage(task);
			}

			// 准备阶段 -- 响应
			else if (v instanceof PrepareResult) {
				this.doSendPrepareResponseMessage(((PrepareResult) v), task);
			}

			// 接受阶段 -- 响应
			else if (v instanceof Proposal && ((Proposal) v).getStep() == 2) {
				this.doSendAcceptMessage(task);
			}

			// 接受阶段 -- 响应
			else if (v instanceof AcceptResult) {
				this.doSendAcceptResponseMessage(((AcceptResult) v), task);
			}

			/*
			 * 100ms 的延迟
			 */
			try {
				Thread.sleep(100);
			} catch (InterruptedException e) {
			}
		}
	}

	/**
	 * 封装的任务
	 * 
	 * @author 618lf
	 */
	class Task {
		Object v;
	}
}
